[SPARK-58089][SQL] Push variant extractions through Aggregate/Sort/Join#57190
Open
qlong wants to merge 1 commit into
Open
[SPARK-58089][SQL] Push variant extractions through Aggregate/Sort/Join#57190qlong wants to merge 1 commit into
qlong wants to merge 1 commit into
Conversation
e26c0de to
16b2f4b
Compare
`PushVariantIntoScan` (v1) and `V2ScanRelationPushDown` (v2)
both rely on `PhysicalOperation`, which collapses only a
contiguous `Project`/`Filter` chain and stops at `Aggregate`,
`Sort`, and `Join`. As a result, `variant_get` expressions
embedded in aggregate function arguments, sort-order
expressions, or join conditions are invisible to the pushdown
rules, and the full variant column is read raw instead of
being shredded to the requested typed fields.
When an `Aggregate` or `Sort` sits above a `Join`, the
barrier compounds: even a `Project` hoisted above the join
tree is still unreachable from the scan side.
Example queries that fail to shred without this fix:
-- Aggregate: extraction in agg arg / GROUP BY
SELECT AVG(variant_get(data, '$.qty', 'double'))
FROM store_sales
GROUP BY variant_get(data, '$.id', 'string')
-- Join: extraction in ON condition
SELECT ss.k
FROM store_sales ss
JOIN date_dim d
ON ss.date_sk = variant_get(d.data, '$.sk', 'int')
-- Aggregate over Join (TPC-DS Q26 shape)
SELECT AVG(variant_get(ss.data, '$.qty', 'double'))
FROM store_sales ss
JOIN date_dim d ON ss.date_sk = d.date_sk
JOIN store st ON ss.store_sk = st.store_sk
GROUP BY variant_get(ss.data, '$.id', 'string')
In all cases the scan emits `data:struct<0:variant>` (full
blob) instead of `data:struct<0:double, 1:string>`.
Introduce a new optimizer rule `PullOutVariantExtractions`,
registered as the first rule in `earlyScanPushDownRules`
(before `SchemaPruning`), which hoists `variant_get` calls
out of `Aggregate`, `Sort`, and `Join` into a `Project`
directly below the operator so the downstream pushdown rules
can see them.
- **Aggregate**: hoist extractions from aggregate function
arguments into a `Project` below the `Aggregate`. The
`Aggregate` defines its own output so the raw variant
column is not passed through.
- **Sort / Join**: match the `Project` sitting directly above
the `Sort`/`Join` (its `references` give the live-above
set) and drop any variant column no longer referenced once
its extraction is hoisted.
- **Push-through-Join**: hoisting above a `Join` is not
enough because `PhysicalOperation` stops there. The new
`pushSideAliases` helper recursively routes each hoisted
`_ve` alias down through the join tree to the side whose
output owns the referenced attribute, landing it in a
`Project` directly above the scan. Handles any depth of
chained joins in one pass.for review
16b2f4b to
ef2a92e
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Introduce a new optimizer rule
PullOutVariantExtractionsthat hoistsvariant_get/Cast(variant)extractions out of three operator typesthat the existing
PushVariantIntoScan/V2ScanRelationPushDownrulescannot see through:
Aggregate function arguments – e.g.
max(variant_get(v, '$.price', 'int')):the extraction is moved into a
Projectdirectly below theAggregateand theaggregate references the resulting alias. The bare variant column is suppressed
unless it is also needed raw (e.g. as a
GROUP BYkey), so no redundantfull-variant slot is generated.
Sort order keys – e.g.
ORDER BY variant_get(v, '$.price', 'int'):matched as
Project → Sort; the extraction is hoisted below theSortandthe original
Projectis reproduced to prevent the alias from leaking intothe output.
Join conditions and projections above joins – matched as
Project → Join;extractions in both the join condition and the outer
Projectare routed tothe owning join side. A
pushSideAliaseshelper then pushes the aliasesthrough any depth of chained joins so they land in a
Projectdirectlyabove the scan (where
PhysicalOperationcollapses them with the scan, makingthem visible to the pushdown). This is necessary because
PhysicalOperationstops at a
Joinnode.A
Sortsitting over aJoinis handled by fusing the two cases: theorder-key aliases are pushed through the join tree, not left in a
Projectabove it.
The rule is gated by a new internal config
spark.sql.variant.pushVariantIntoScan.pullOutExtractions(defaulttrue)and is a no-op unless
spark.sql.variant.pushVariantIntoScanis also enabled.Non-variant plans are untouched.
The rule is registered as the first rule in
SparkOptimizer.earlyScanPushDownRules,before
SchemaPruningand the V2 scan pushdown rules.Why are the changes needed?
Before this change, a
variant_getinside an aggregate function argument, sort key,or join condition caused the whole variant column to be read raw (or shredded with a
redundant full-variant slot). For example:
read the entire v column even though only the price field was needed.
After this change, Spark shreds only the requested typed fields, avoiding the
full-variant I/O.
The change improves query performance for variant referenced in aggregrate, join, sort.
Does this PR introduce any user-facing change?
No
How was this patch tested?
Added new units. Also run correctness tests against some known workload.
Performance result
The test run against tpc-ds dataset (SF=5). Run B is with the new pullout rule enabled.
Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Code